package mapreduce

import (
	"encoding/json"
	"io"
	"log"
	"os"
	"sort"
)

type ByKey []KeyValue

func (a ByKey) Len() int 			{return len(a)}
func (a ByKey) Swap(i, j int)		{a[i], a[j] = a[j], a[i]}
func (a ByKey) Less(i, j int) bool	{return a[i].Key < a[j].Key}

func DoReduce(
	jobName string,
	reduceTaskNumber int,
	outFile string,
	nMap int,
	reduceF func(key string, values []string) string,
){
	//已按照Key值排序的pairs
	var sortedPairs []KeyValue

	//每一个map任务都会产生一个与此reduce相对应的输出文件
	//fmt.Println(nMap)
	//读取每一个map为此reduce产生的文件
	for i:=0; i<nMap; i++ {
		interFilename := reduceName(jobName,i, reduceTaskNumber)
		interFile, err := os.Open(interFilename)
		if err != nil {
			log.Fatal(err)
		}
		dec := json.NewDecoder(interFile)
		for {
			var kv KeyValue
			if err := dec.Decode(&kv); err ==io.EOF {
				break
			}else if err != nil {
				log.Fatal(err)
			}
			sortedPairs = append(sortedPairs, kv)
		}
		interFile.Close()
	}
	sort.Sort(ByKey(sortedPairs))
	//reduceF输入数据
	var key string
	var values []string
	var outPairs []KeyValue
	for i, pair := range sortedPairs  {
		if i==0 {
			key = pair.Key
		}

		if key == pair.Key {
			values = append(values, pair.Value)
		}else{
			result := reduceF(key, values)
			outPair := KeyValue{key, result}
			outPairs = append(outPairs, outPair)
			key = pair.Key
			values = values[:0]
			values = append(values, pair.Value)
		}

		if i==len(sortedPairs)-1 {
			result := reduceF(key, values)
			outPair := KeyValue{key, result}
			outPairs = append(outPairs, outPair)
		}
	}

	//encode pairs into disk
	file, err := os.Create(outFile)
	if err != nil {
		log.Fatal(err)
	}
	enc := json.NewEncoder(file)
	for _, pair := range outPairs {
		enc.Encode(pair)
	}
	file.Close()
}
